package org.example.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author xw
 * @date 2021/6/27 7:01 下午
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter,Runnable {

    private static final Logger log = LoggerFactory.getLogger(TPMonitorFilter.class);
    /** 定义线程池使用的阈值 */
    private final Map<String, Vector<KeyVal>> TP_MAPS = new ConcurrentHashMap<>();

    public TPMonitorFilter(){
        // 每个5秒钟打印tp90, tp99使用情况
        Executors.newSingleThreadScheduledExecutor()
                .scheduleWithFixedDelay(this, 1, 5, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        String methodName = invocation.getMethodName();
        try {
            return invoker.invoke(invocation);
        } finally {
            long endTime = System.currentTimeMillis();
            long times = endTime - startTime;
//            log.info("invoke time: {} 毫秒" , times);

            KeyVal keyVal = new KeyVal(endTime, times);
            Vector<KeyVal> keyVals = TP_MAPS.get(methodName);
            if (keyVals == null){
                Vector<KeyVal> objects = new Vector<>();
                objects.add(keyVal);
                TP_MAPS.put(methodName, objects);
            }else {
                keyVals.add(keyVal);
            }
        }
    }

    @Override
    public void run() {
        TP_MAPS.forEach((k, v) -> {
            List<KeyVal> collect = v.stream().filter(a -> System.currentTimeMillis() - a.getTime() > 60000).collect(Collectors.toList());
            v.removeAll(collect);
            tp(v, 90, k);
            tp(v, 99, k);
        });
    }

    private void tp(Vector<KeyVal> vals, int t, String methodName){
        int size = vals.size();
        int index = size * t / 100;
        vals.sort(Comparator.comparing(KeyVal::getTimes));
        KeyVal keyVal = vals.get(index);
        log.info("接口： {}, 统计条数： {}，定位 {},  tp{} = {}", methodName, size, index, t, keyVal.getTimes());
    }

}
